package com.shujia.flink.tf

import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time

object Demo3Window {
  def main(args: Array[String]): Unit = {

    //创建flink的运行环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    val ds: DataStream[String] = env.socketTextStream("master", 7777)

    /**
      * 统计最近5秒单词的数量
      *
      */

    val kvDS: DataStream[(String, Int)] = ds.flatMap(_.split(",")).map((_, 1))


    val keyByDS: KeyedStream[(String, Int), String] = kvDS.keyBy(_._1)


    keyByDS
      //滚动窗口
      //.timeWindow(Time.seconds(5))

      //滑动窗口
      //窗口大小10秒每隔两秒一个窗口
      .timeWindow(Time.seconds(10), Time.seconds(2))
      .sum(1)
      .print()


    env.execute()

  }
}
